package com.apes.framework.util.monitor;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apes.framework.config.druid.stat.ApesDruidDataSourceStatLogger;
import com.apes.framework.config.redis.RedisMessageListener;
import com.apes.framework.plugin.cache.Cache;
import com.apes.framework.util.ReflectUtil;
import org.springframework.beans.factory.DisposableBean;

import javax.sql.DataSource;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicLong;

public abstract class AbstractMonitorService<T> implements DisposableBean {
    protected AtomicLong counter = new AtomicLong(0);
    protected Queue<T> queue = new LinkedList<>();
    private Thread thread;
    private volatile boolean stop = false;
    private boolean enable = false;

    protected DruidDataSource druidDataSource;
    private Cache cache;

    public AbstractMonitorService(Cache cache, DataSource dataSource) {
        this.cache = cache;
        Map<String, Boolean> settings = (Map<String, Boolean>) this.cache.get(Constants.MONITOR_SETTING_TOPIC);
        if (settings != null) {
            String topic = getTopic();
            if (settings.containsKey(topic)) {
                boolean enable = settings.get(topic);
                setEnable(enable);
            }
        }
        if (dataSource != null) {
            if (dataSource instanceof DruidDataSource) {
                druidDataSource = (DruidDataSource) dataSource;
            } else {
                druidDataSource = (DruidDataSource) ReflectUtil.getPrivateField(dataSource, "realDataSource");
            }
        }
        if (isEnable()) {
            start();
        }
    }

    @Override
    public void destroy() {
        stop();
    }

    /**
     * 增加一个元索
     */
    public void add(T data) {
        if (!isEnable()) {
            return;
        }
        queue.add(data);
    }

    /**
     * 移除并返问队列头部的元素
     **/
    public T poll() {
        if (queue.isEmpty()) {
            return null;
        }
        return queue.poll();
    }

    private Thread createThread() {
        Thread thread = new Thread(() -> {
            while (!stop) {
                T data = poll();
                if (data != null) {
                    try {
                        run(data);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                } else {
                    try {
                        Thread.sleep(1000);
                    } catch (Exception e) {
                    }
                }
            }
        });
        thread.setDaemon(true);
        thread.setName(this.getClass().getSimpleName() + counter.incrementAndGet());
        thread.start();
        return thread;
    }

    public void start() {
        if (druidDataSource != null) {
            druidDataSource.setRemoveAbandoned(true);
            if (druidDataSource.getStatLogger() == null) {
                druidDataSource.setStatLogger(new ApesDruidDataSourceStatLogger());
            }
        }
        if (stop || thread == null) {
            thread = createThread();
            stop = false;
        }
    }

    public void stop() {
        if (druidDataSource != null) {
            druidDataSource.setRemoveAbandoned(false);
            druidDataSource.setStatLogger(null);
        }
        stop = true;
        if (thread != null) {
            thread.interrupt();
            try {
                thread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public abstract void run(T data);

    public abstract String getTopic();

    @RedisMessageListener(topic = Constants.MONITOR_SETTING_TOPIC)
    public void subscribe(String message) {
        JSONObject jo = JSON.parseObject(message);
        boolean enable = false;
        String topic = getTopic();
        if (jo.containsKey(topic)) {
            enable = jo.getBoolean(topic);
        }
        setEnable(enable);
        if (isEnable()) {
            start();
        } else {
            stop();
        }
        if (jo.containsKey(Constants.MONITOR_DATA_DELETE)) {
            boolean deleted = jo.getBoolean(Constants.MONITOR_DATA_DELETE);
            if (deleted) {
                deleteAllData();
            }
        }
    }

    protected abstract void deleteAllData();

    public boolean isEnable() {
        return enable;
    }

    public void setEnable(boolean enable) {
        this.enable = enable;
    }
}
